package com.simple.datax;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.xmlbeans.XmlException;

import cn.gov.chinaport.tcs.flow.TcsFlow201Document;
import cn.gov.chinaport.tcs.flow.TcsFlowType;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.api.Connector;
import com.simple.datax.api.ConnectorException;
import com.simple.datax.conn.impl.ConnectorFactory;
import com.simple.datax.xml.tcs.MessageHeadDocument.MessageHead;
import com.simple.datax.xml.tcs.TCS101MessageDocument;
import com.simple.datax.xml.tcs.TCS101MessageDocument.TCS101Message;
import com.simple.datax.xml.tcs.XMLDATA;

public class TCSLNSDataSimulator {
	private static final Logger logger = LoggerFactory
			.getLogger(TCSLNSDataSimulator.class);
	
//	private List<Connector> simDataConnList;
//	private List<Connector> resultDataConnList;
	private int repeat;
	private static String encoding="utf-8";
	private TCS101MessageDocument catdoc;
	private TcsFlowType ecFlow;
	private MessageHead msgHead;
	private TCS101Message message;
	private TcsFlow201Document ecDoc;
	private String file;
	
	private long recvTotal;
	private long sendTotal;
	private long start;
	private static final Pattern PATTERN_MSG_ID = Pattern.compile("<MessageId>(.*)</MessageId>");
	
	public static void main(String[] args) {
		Properties prop = new Properties();
		FileInputStream fis = null;
		Map<String, String> prodDataConnList = null;
		Map<String, String> consDataConnList = null;
		String template = null;
		String operation = null;
		String speed = null;
		String messageid_start = null;
		if (args.length != 1) {
			logger.error("[Usage] java LNSDataSimulator Operation:");
			logger.error(" Operation option:");
			logger.error(" 1: Send message to specified target and Receive from other target.");
			logger.error(" 2: Receive message from target Only");
			return;
		}
		operation = args[0];
		if (!operation.equals("1") && !operation.equals("2")) {
			logger.error("Command parameter error. Input 1 or 2.");
			return;
		}
		try {
			fis = new FileInputStream("tcslnsdata.properties");
			prop.load(fis);
			
			prodDataConnList = getRecvConnectorType(prop);
			consDataConnList = getSendConnectorType(prop);
			
			template = prop.getProperty("template");
			if (template == null || template.trim().isEmpty()) {
				logger.error("Please set message template file path.");
				return;
			}
			
			speed = prop.getProperty("simulator_speed");
			if (speed == null || speed.trim().isEmpty()) {
				logger.error("Please set speed.");
				return;
			}
			
			messageid_start = prop.getProperty("messageid_start");
			if (messageid_start == null || messageid_start.trim().isEmpty()) {
				logger.error("Please set messageid_start.");
				return;
			}
			encoding = prop.getProperty("encoding");
			if (encoding == null || encoding.trim().isEmpty()) {
				encoding = "utf-8";
			}
			
		} catch (Exception ex) {
			logger.error(ex);
			return;
		} finally {
			try {
				if (fis != null) {
					fis.close();
				}
			} catch (IOException e) {
				logger.error(e);
			}
		}
        
		TCSLNSDataSimulator sim = new TCSLNSDataSimulator(template, Integer.valueOf(speed), Long.valueOf(messageid_start));
		if (operation.equals("1")) {
			sim.startConsumerThread(consDataConnList);
			sim.startProductThread(prodDataConnList);
		} else {
			sim.startConsumerThread(consDataConnList);
		}
	}
	
	private static Map<String, String> getRecvConnectorType(Properties prop) {
		Map<String, String> ret = new HashMap<String, String>();
		if (prop.getProperty("MSMQ_RECV") != null) {
			ret.put("MSMQ", prop.getProperty("MSMQ_RECV"));
		}
		if (prop.getProperty("IBMMQ_RECV") != null) {
			ret.put("IBMMQ", prop.getProperty("IBMMQ_RECV"));
		}
		if (prop.getProperty("FTP_RECV") != null) {
			ret.put("FTP", prop.getProperty("FTP_RECV"));
		}
		if (prop.getProperty("SFTP_RECV") != null) {
			ret.put("SFTP", prop.getProperty("SFTP_RECV"));
		}
		if (prop.getProperty("FTP_RECV") != null) {
			ret.put("FTP", prop.getProperty("FTP_RECV"));
		}
		if (prop.getProperty("FTPS_RECV") != null) {
			ret.put("FTPS", prop.getProperty("FTPS_RECV"));
		}
		if (prop.getProperty("ACTIVEMQ_RECV") != null) {
			ret.put("ACTIVEMQ", prop.getProperty("ACTIVEMQ_RECV"));
		}
		return ret;
	}
	
	private static Map<String, String> getSendConnectorType(Properties prop) {
		Map<String, String> ret = new HashMap<String, String>();
		if (prop.getProperty("MSMQ_SEND") != null) {
			ret.put("MSMQ", prop.getProperty("MSMQ_SEND"));
		}
		if (prop.getProperty("IBMMQ_SEND") != null) {
			ret.put("IBMMQ", prop.getProperty("IBMMQ_SEND"));
		}
		if (prop.getProperty("FTP_SEND") != null) {
			ret.put("FTP", prop.getProperty("FTP_SEND"));
		}
		if (prop.getProperty("SFTP_SEND") != null) {
			ret.put("SFTP", prop.getProperty("SFTP_SEND"));
		}
		if (prop.getProperty("FTP_SEND") != null) {
			ret.put("FTP", prop.getProperty("FTP_SEND"));
		}
		if (prop.getProperty("FTPS_SEND") != null) {
			ret.put("FTPS", prop.getProperty("FTPS_SEND"));
		}
		if (prop.getProperty("ACTIVEMQ_SEND") != null) {
			ret.put("ACTIVEMQ", prop.getProperty("ACTIVEMQ_SEND"));
		}
		return ret;
	}
	
	public TCSLNSDataSimulator(String file, int repeat, long start) {
		this.repeat = repeat;
		this.file = file;
		this.start = start;
	}
	
	/**
	 * 初始化所有指定连接器
	 * @param connector
	 */
	private List<Connector> initConnector(Map<String, String> connector) {
		if (connector == null) {
			return null;
		}
		List<Connector> connectorList = new ArrayList<Connector>(connector.keySet().size());
		String connPara = null;
		for (String connType:connector.keySet()) {
			connPara = connector.get(connType);
			connPara = connType + ConnectorFactory.KEY_SPLITTER + connPara;
			try {
				connectorList.add(ConnectorFactory.getInstance().makeObject(connPara));
			} catch (Exception e) {
				logger.error("Fail to initial connector - " + connPara, e);
			}
		}
		return connectorList;
	}
	
	public String getMessageContent(long id) {
		String tmp = String.format("%1$09d", id);
		String taskId = ecFlow.getTaskId();
		
		if (taskId != null) {
			if (taskId.length()<=13) {
				tmp = taskId + tmp;
			} else {
				tmp = taskId.substring(0, 13) + tmp;
			}
		}
		msgHead.setMessageId(tmp);
		ecFlow.setTaskId(tmp);
		ecFlow.setMessageId(tmp);
		try {
			message.setMessageBody(XMLDATA.Factory.parse(ecDoc.toString()));
		} catch (XmlException e) {
			e.printStackTrace();
		}
		return catdoc.toString();
	}
	
	public static String readFileByLines(String fileName) {

		StringBuffer sb = new StringBuffer("");
		BufferedReader reader = null;
		try {
			// System.out.println("以行为单位读取文件内容，一次读一整行：");
			FileInputStream fis = new FileInputStream(fileName);
			InputStreamReader isr = new InputStreamReader(fis, "UTF-8");
			reader = new BufferedReader(isr);
			String tempString = null;
			int line = 1;
			// 一次读入一行，直到读入null为文件结束
			while ((tempString = reader.readLine()) != null) {
				// 显示行号
				// System.out.println("line " + line + ": " + tempString);
				line++;

				sb.append(tempString).append("\r\n");
			}
			reader.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (reader != null) {
				try {
					reader.close();
				} catch (IOException e1) {
				}
			}
		}

		return sb.toString();
	}
	
	/**
	 * 从报文中读取内容
	 * @param file
	 * @return
	 */
	private void readMessageFromFile() {
		if (file == null) {
			return;
		}
		try {
			String xmlfile = readFileByLines(file);
			if (!xmlfile.startsWith("<")) {
				int beginIndex = xmlfile.indexOf("<");
				xmlfile = xmlfile.substring(beginIndex);
//				System.out.println(beginIndex);
			}
			
//			BufferedReader bufReader=new BufferedReader(new InputStreamReader(new FileInputStream(file), getEncoding()));
			catdoc = TCS101MessageDocument.Factory.parse(xmlfile);
			message = catdoc.getTCS101Message();
			msgHead = message.getMessageHead();
			ecDoc = TcsFlow201Document.Factory.parse(message.getMessageBody().xmlText());
			ecFlow = ecDoc.getTcsFlow201().getTcsFlow();
		} catch (Exception ex) {
			logger.error("Failed to read message from " + file, ex);
		}
	}

	public String getEncoding() {
		return encoding;
	}
	
	public void setEncoding(String encoding) {
		this.encoding = encoding;
	}

	/**
	 * 开始接受结果数据
	 * @param connector
	 */
	public void startConsumerThread(Map<String, String> connector) {
		Thread thread = new Thread(new DataConsumer(initConnector(connector)));
		thread.start();
	}
	
	/**
	 * 开始模拟发送数据
	 * @param connector
	 */
	public void startProductThread(Map<String, String> connector) {
		Thread thread = new Thread(new DataProductor(initConnector(connector)));
		thread.start();
	}

	/**
	 * 模拟数据，并发送至指定连接器
	 * @author Jin Yi
	 *
	 */
	public class DataProductor implements Runnable {

		private List<Connector> connectorList;
		long id = start;
		public DataProductor(List<Connector> connectorList) {
			this.connectorList = connectorList;
		}
		
		@Override
		public void run() {
			// Read message from file;
			readMessageFromFile();
			//去掉循环发送 2013-03-04
			//while (true) {   
//				logger.info("");
				sendMessage();
				try {
					  Thread.sleep(1000);
				} catch (Exception ex) {
					
				}
			//}
			
		}
		
		private void sendMessage() {
			
			String message = null;
			for (int i=0; i<repeat; i++) {
				for(Connector conn:connectorList) {
					message = getMessageContent(id);
					logger.info("Send: " + message);
					SimpleMessage sm = new SimpleMessage();
					sm.setContent(message);
					sm.setRecvRawMsgid(String.valueOf(id));
					id++;
					conn.send(sm);
					logger.debug(message);
					logger.info("======== Total Send " + ++sendTotal);
				}
			}
		}
		
	}
	
	/**
	 * 从指定连接器中读取处理结果
	 * @author Jin Yi
	 *
	 */
	public class DataConsumer implements Runnable {

		private List<Connector> connectorList;
//		private Set<String> set = new HashSet<String>(20000);
		
		public DataConsumer(List<Connector> connectorList) {
			this.connectorList = connectorList;
		}
		
		@Override
		public void run() {
			
			while (true) {
				getMessage();
				try {
					  Thread.sleep(1000);
				} catch (Exception ex) {
					
				}
			}
		}
		
		private void getMessage() {
			String str = null;
			for(Connector conn:connectorList) {
				try {
					str = conn.getMessage();
//					logger.info(" &&**&&* " + str);
					while (str != null) {
						getMessageId(str);
						logger.info("======== Total Recv " + ++recvTotal);
						str = conn.getMessage();
					}
				} catch (ConnectorException e) {
					logger.error("Failed to get message.", e);
				}
			}
		}
		
		private void getMessageId(String message) {
			Matcher match = PATTERN_MSG_ID.matcher(message);
			if (match.find()) {
				if (match.groupCount()==1) {
					String tmp = match.group(1);
//					logger.info(message);
					logger.info(" &&&&&&&& "+ tmp);
//					if (set.contains(tmp)) {
//						logger.error(" ******** Existed - "+ tmp);
//					} else {
//						set.add(tmp);
//					}
				} else {
					logger.error(" ******** Wrong Id - "+match.group(0));
				}
			}
			match = null;
		}
	}
}
